#include "rm_file_handle.h"

#include <assert.h>
#include <cstring> // For memcpy
#include <stdexcept> // For std::runtime_error

// ¸¨Öúº¯Êý£º½«Ò³Ãæ±ê¼ÇÎªÔà²¢È¡Ïû¹Ì¶¨
static void MarkDirtyAndUnpin(BufferPoolManager *bpm, Page *page, bool is_dirty) {
    if (!bpm->UnpinPage(page->GetPageId(), is_dirty)) {
        // Ó¦¸Ã²»»á·¢Éú£¬ÒòÎªÎÒÃÇÍ¨³£ÔÚ pin_count > 0 Ê±µ÷ÓÃ UnpinPage
        // »òÕßÔÚ pin_count == 0 Ê±£¬UnpinPage ÄÚ²¿»á´¦Àí
        // ÕâÀï¿ÉÒÔÅ×³ö¸ü¾ßÌåµÄ´íÎó»ò¼ÇÂ¼ÈÕÖ¾
        throw InternalError("Failed to unpin page " + std::to_string(page->GetPageId().page_no));
    }
}

/**
 * @brief ÓÉRidµÃµ½Ö¸ÏòRmRecordµÄÖ¸Õë
 *
 * @param rid Ö¸¶¨¼ÇÂ¼ËùÔÚµÄÎ»ÖÃ
 * @return std::unique_ptr<RmRecord>
 */
std::unique_ptr<RmRecord> RmFileHandle::get_record(const Rid &rid, Context *context) const {
    // Todo:
    // 1. »ñÈ¡Ö¸¶¨¼ÇÂ¼ËùÔÚµÄpage handle
    // 2. ³õÊ¼»¯Ò»¸öÖ¸ÏòRmRecordµÄÖ¸Õë£¨¸³ÖµÆäÄÚ²¿µÄdataºÍsize£©

    // »ñÈ¡Ö¸¶¨¼ÇÂ¼ËùÔÚµÄpage handle
    RmPageHandle page_handle = fetch_page_handle(rid.page_no);

    // ¼ì²é¼ÇÂ¼ÊÇ·ñ´æÔÚ
    if (!Bitmap::is_set(page_handle.bitmap, rid.slot_no)) {
        // ÊÍ·ÅÒ³Ãæ
        MarkDirtyAndUnpin(buffer_pool_manager_, page_handle.page, false);
        throw RecordNotFoundError(rid.page_no, rid.slot_no);
    }

    // ³õÊ¼»¯Ò»¸öÖ¸ÏòRmRecordµÄÖ¸Õë
    std::unique_ptr<RmRecord> record = std::make_unique<RmRecord>(file_hdr_.record_size);
    char *slot_data = page_handle.get_slot(rid.slot_no);
    record->SetData(slot_data); // ¿½±´Êý¾Ýµ½RmRecordµÄÄÚ²¿»º³åÇø

    // ÊÍ·ÅÒ³Ãæ
    MarkDirtyAndUnpin(buffer_pool_manager_, page_handle.page, false);

    return record;
}

/**
 * @brief ÔÚ¸Ã¼ÇÂ¼ÎÄ¼þ£¨RmFileHandle£©ÖÐ²åÈëÒ»Ìõ¼ÇÂ¼
 *
 * @param buf Òª²åÈëµÄÊý¾ÝµÄµØÖ·
 * @return Rid ²åÈë¼ÇÂ¼µÄÎ»ÖÃ
 */
Rid RmFileHandle::insert_record(char *buf, Context *context) {
    // Todo:
    // 1. »ñÈ¡µ±Ç°Î´ÂúµÄpage handle
    // 2. ÔÚpage handleÖÐÕÒµ½¿ÕÏÐslotÎ»ÖÃ
    // 3. ½«buf¸´ÖÆµ½¿ÕÏÐslotÎ»ÖÃ
    // 4. ¸üÐÂpage_handle.page_hdrÖÐµÄÊý¾Ý½á¹¹
    // ×¢Òâ¿¼ÂÇ²åÈëÒ»Ìõ¼ÇÂ¼ºóÒ³ÃæÒÑÂúµÄÇé¿ö£¬ÐèÒª¸üÐÂfile_hdr_.first_free_page_no

    // 1. »ñÈ¡µ±Ç°Î´ÂúµÄpage handle
    RmPageHandle page_handle = create_page_handle(); // create_page_handle »á pin Ò³Ãæ

    // 2. ÔÚpage handleÖÐÕÒµ½¿ÕÏÐslotÎ»ÖÃ
    int slot_no = Bitmap::first_bit(false, page_handle.bitmap, file_hdr_.num_records_per_page);
    if (slot_no == file_hdr_.num_records_per_page) {
        // ÀíÂÛÉÏ create_page_handle Ó¦¸Ã·µ»ØÒ»¸öÓÐ¿ÕÏÐ slot µÄÒ³Ãæ
        // Èç¹û³öÏÖÕâÖÖÇé¿ö£¬¿ÉÄÜÊÇÂß¼­´íÎó»ò²¢·¢ÎÊÌâ£¬ÕâÀïÅ×³öÄÚ²¿´íÎó
        MarkDirtyAndUnpin(buffer_pool_manager_, page_handle.page, false);
        throw InternalError("No free slot found in a newly created/fetched free page.");
    }

    // 3. ½«buf¸´ÖÆµ½¿ÕÏÐslotÎ»ÖÃ
    char *slot_data = page_handle.get_slot(slot_no);
    memcpy(slot_data, buf, file_hdr_.record_size);

    // 4. ¸üÐÂpage_handle.page_hdrÖÐµÄÊý¾Ý½á¹¹
    Bitmap::set(page_handle.bitmap, slot_no); // ½«¶ÔÓ¦ slot ÖÃÎªÒÑÕ¼ÓÃ
    page_handle.page_hdr->num_records++;      // Ò³Ãæ¼ÇÂ¼Êý¼Ó1

    Rid inserted_rid = {page_handle.page->GetPageId().page_no, slot_no};

    // ×¢Òâ¿¼ÂÇ²åÈëÒ»Ìõ¼ÇÂ¼ºóÒ³ÃæÒÑÂúµÄÇé¿ö£¬ÐèÒª¸üÐÂfile_hdr_.first_free_page_no
    if (page_handle.page_hdr->num_records == file_hdr_.num_records_per_page) {
        // Èç¹ûÒ³ÃæÒÑÂú£¬½«Æä´Ó¿ÕÏÐÁ´±íÖÐÒÆ³ý
        // ´ËÊ±¸ÃÒ³ÃæµÄ next_free_page_no Ó¦¸Ã±»ÉèÖÃÎª RM_NO_PAGE
        // µ«ÎªÁË±£³ÖÁ´±í½á¹¹£¬ÎÒÃÇ¸üÐÂ file_hdr_.first_free_page_no
        // Ö¸Ïò¸ÃÒ³ÃæµÄÏÂÒ»¸ö¿ÕÏÐÒ³Ãæ£¨Èç¹û´æÔÚµÄ»°£©£¬È»ºó½«¸ÃÒ³µÄ next_free_page_no ÉèÎª RM_NO_PAGE
        // ÕâÀïµÄÂß¼­ÐèÒªºÍ release_page_handle ÅäºÏ
        if (file_hdr_.first_free_page_no == page_handle.page->GetPageId().page_no) {
            file_hdr_.first_free_page_no = page_handle.page_hdr->next_free_page_no;
        }
        page_handle.page_hdr->next_free_page_no = RM_NO_PAGE; // ±ê¼Ç¸ÃÒ³ÒÑÂú
    }

    // ±ê¼ÇÒ³ÃæÎªÔà²¢È¡Ïû¹Ì¶¨
    MarkDirtyAndUnpin(buffer_pool_manager_, page_handle.page, true);

    return inserted_rid;
}

/**
 * @brief ÔÚ¸Ã¼ÇÂ¼ÎÄ¼þ£¨RmFileHandle£©ÖÐÉ¾³ýÒ»ÌõÖ¸¶¨Î»ÖÃµÄ¼ÇÂ¼
 *
 * @param rid ÒªÉ¾³ýµÄ¼ÇÂ¼ËùÔÚµÄÖ¸¶¨Î»ÖÃ
 */
void RmFileHandle::delete_record(const Rid &rid, Context *context) {
    // Todo:
    // 1. »ñÈ¡Ö¸¶¨¼ÇÂ¼ËùÔÚµÄpage handle
    // 2. ¸üÐÂpage_handle.page_hdrÖÐµÄÊý¾Ý½á¹¹
    // ×¢Òâ¿¼ÂÇÉ¾³ýÒ»Ìõ¼ÇÂ¼ºóÒ³ÃæÎ´ÂúµÄÇé¿ö£¬ÐèÒªµ÷ÓÃrelease_page_handle()

    // 1. »ñÈ¡Ö¸¶¨¼ÇÂ¼ËùÔÚµÄpage handle
    RmPageHandle page_handle = fetch_page_handle(rid.page_no);

    // ¼ì²é¼ÇÂ¼ÊÇ·ñ´æÔÚ
    if (!Bitmap::is_set(page_handle.bitmap, rid.slot_no)) {
        MarkDirtyAndUnpin(buffer_pool_manager_, page_handle.page, false);
        throw RecordNotFoundError(rid.page_no, rid.slot_no);
    }

    // ¼ì²éÒ³ÃæÊÇ·ñÒÑÂú£¨É¾³ýÇ°£©
    bool was_full = (page_handle.page_hdr->num_records == file_hdr_.num_records_per_page);

    // 2. ¸üÐÂpage_handle.page_hdrÖÐµÄÊý¾Ý½á¹¹
    Bitmap::reset(page_handle.bitmap, rid.slot_no); // ½«¶ÔÓ¦ slot ÖÃÎªÎ´Õ¼ÓÃ
    page_handle.page_hdr->num_records--;            // Ò³Ãæ¼ÇÂ¼Êý¼õ1

    // ×¢Òâ¿¼ÂÇÉ¾³ýÒ»Ìõ¼ÇÂ¼ºóÒ³ÃæÎ´ÂúµÄÇé¿ö£¬ÐèÒªµ÷ÓÃrelease_page_handle()
    // Èç¹ûÒ³ÃæÖ®Ç°ÊÇÂúµÄ£¬ÏÖÔÚ²»Âú£¬Ôò½«ÆäÌí¼Óµ½¿ÕÏÐÁ´±í
    if (was_full && page_handle.page_hdr->num_records < file_hdr_.num_records_per_page) {
        release_page_handle(page_handle);
    }

    // ±ê¼ÇÒ³ÃæÎªÔà²¢È¡Ïû¹Ì¶¨
    MarkDirtyAndUnpin(buffer_pool_manager_, page_handle.page, true);
}

/**
 * @brief ¸üÐÂÖ¸¶¨Î»ÖÃµÄ¼ÇÂ¼
 *
 * @param rid Ö¸¶¨Î»ÖÃµÄ¼ÇÂ¼
 * @param buf ÐÂ¼ÇÂ¼µÄÊý¾ÝµÄµØÖ·
 */
void RmFileHandle::update_record(const Rid &rid, char *buf, Context *context) {
    // Todo:
    // 1. »ñÈ¡Ö¸¶¨¼ÇÂ¼ËùÔÚµÄpage handle
    // 2. ¸üÐÂ¼ÇÂ¼

    // 1. »ñÈ¡Ö¸¶¨¼ÇÂ¼ËùÔÚµÄpage handle
    RmPageHandle page_handle = fetch_page_handle(rid.page_no);

    // ¼ì²é¼ÇÂ¼ÊÇ·ñ´æÔÚ
    if (!Bitmap::is_set(page_handle.bitmap, rid.slot_no)) {
        MarkDirtyAndUnpin(buffer_pool_manager_, page_handle.page, false);
        throw RecordNotFoundError(rid.page_no, rid.slot_no);
    }

    // 2. ¸üÐÂ¼ÇÂ¼
    char *slot_data = page_handle.get_slot(rid.slot_no);
    memcpy(slot_data, buf, file_hdr_.record_size);

    // ±ê¼ÇÒ³ÃæÎªÔà²¢È¡Ïû¹Ì¶¨
    MarkDirtyAndUnpin(buffer_pool_manager_, page_handle.page, true);
}

/** -- ÒÔÏÂÎª¸¨Öúº¯Êý -- */
/**
 * @brief »ñÈ¡Ö¸¶¨Ò³Ãæ±àºÅµÄpage handle
 *
 * @param page_no Òª»ñÈ¡µÄÒ³Ãæ±àºÅ
 * @return RmPageHandle ·µ»Ø¸øÉÏ²ãµÄpage_handle
 * @note pin the page, remember to unpin it outside!
 */
RmPageHandle RmFileHandle::fetch_page_handle(int page_no) const {
    // Todo:
    // Ê¹ÓÃ»º³å³Ø»ñÈ¡Ö¸¶¨Ò³Ãæ£¬²¢Éú³Épage_handle·µ»Ø¸øÉÏ²ã
    // if page_no is invalid, throw PageNotExistError exception

    if (page_no < 0 || page_no >= file_hdr_.num_pages) {
        throw PageNotExistError(disk_manager_->GetFileName(fd_), page_no);
    }

    PageId page_id = {fd_, page_no};
    Page *page = buffer_pool_manager_->FetchPage(page_id);
    if (page == nullptr) {
        // FetchPage ·µ»Ø nullptr Í¨³£ÒâÎ¶×Å»º³å³ØÒÑÂúÇÒÃ»ÓÐ¿ÉÌÔÌ­µÄÒ³Ãæ
        // »òÕßÒ³Ãæ²»´æÔÚÓÚ´ÅÅÌ£¨Èç¹û DiskManager Ò²»á·µ»Ø nullptr£©
        throw InternalError("Failed to fetch page " + std::to_string(page_no) + " from buffer pool.");
    }
    return RmPageHandle(&file_hdr_, page);
}

/**
 * @brief ´´½¨Ò»¸öÐÂµÄpage handle
 *
 * @return RmPageHandle
 */
RmPageHandle RmFileHandle::create_new_page_handle() {
    // Todo:
    // 1.Ê¹ÓÃ»º³å³ØÀ´´´½¨Ò»¸öÐÂpage
    // 2.¸üÐÂpage handleÖÐµÄÏà¹ØÐÅÏ¢
    // 3.¸üÐÂfile_hdr_

    PageId new_page_id;
    new_page_id.fd = fd_; // ÐÂÒ³ÃæµÄÎÄ¼þÃèÊö·ûÓëµ±Ç°ÎÄ¼þ¾ä±úÏàÍ¬
    Page *new_page = buffer_pool_manager_->NewPage(&new_page_id);
    if (new_page == nullptr) {
        throw InternalError("Failed to create a new page in buffer pool.");
    }

    // ¸üÐÂ file_hdr_ ÖÐµÄÒ³ÃæÊýÁ¿
    file_hdr_.num_pages++;

    // ³õÊ¼»¯ÐÂµÄ RmPageHandle
    RmPageHandle new_page_handle(&file_hdr_, new_page);

    // ³õÊ¼»¯ RmPageHdr
    new_page_handle.page_hdr->next_free_page_no = RM_NO_PAGE; // ³õÊ¼Ê±±ê¼ÇÎªÒÑÂú£¨Ã»ÓÐÏÂÒ»¸ö¿ÕÏÐÒ³£©
    new_page_handle.page_hdr->num_records = 0;                 // ³õÊ¼¼ÇÂ¼ÊýÎª0

    // ³õÊ¼»¯ bitmap ÎªÈ« 0 (ËùÓÐ slot ¿ÕÏÐ)
    Bitmap::init(new_page_handle.bitmap, file_hdr_.bitmap_size);

    // Èç¹ûµ±Ç°Ã»ÓÐ¿ÕÏÐÒ³Á´±í£¬»òÕßÐÂ´´½¨µÄÒ³ÃæÊÇµÚÒ»¸ö¿ÕÏÐÒ³£¬¸üÐÂ first_free_page_no
    if (file_hdr_.first_free_page_no == RM_NO_PAGE) {
        file_hdr_.first_free_page_no = new_page_id.page_no;
    } else {
        // ·ñÔò£¬½«ÐÂÒ³ÃæÌí¼Óµ½¿ÕÏÐÁ´±íµÄÄ©Î²£¨»òÕßÍ·²¿£¬È¡¾öÓÚ²ßÂÔ£©
        // ÕâÀïÎªÁË¼òµ¥£¬ÎÒÃÇ¼ÙÉèÐÂ´´½¨µÄÒ³Ãæ»á³ÉÎªÐÂµÄ¿ÕÏÐÁ´±íµÄÍ·²¿£¨Èç¹ûËüÓÐ¿ÕÏÐ²ÛÎ»µÄ»°£©
        // »òÕß£¬¸ü³£¼ûµÄ×ö·¨ÊÇ£¬Èç¹ûËüÓÐ¿ÕÏÐ²ÛÎ»£¬Ëü»á³ÉÎªÐÂµÄ first_free_page_no
        // ²¢ÇÒÖ®Ç°µÄ first_free_page_no »áÍ¨¹ýÆä next_free_page_no Ö¸ÏòËü¡£
        // µ«ÒòÎª create_new_page_handle ×ÜÊÇ´´½¨Ò»¸ö¡°¿Õ¡±µÄÐÂÒ³Ãæ£¬
        // ËüµÄ num_records Îª 0£¬ËùÒÔËü±¾Éí¾ÍÊÇÒ»¸ö¿ÕÏÐÒ³¡£
        // ÎÒÃÇ¿ÉÒÔ½«Æä²åÈëµ½¿ÕÏÐÁ´±íµÄÍ·²¿£¬ÈÃËü³ÉÎªÐÂµÄ first_free_page_no¡£
        // ÕâÖÖÇé¿öÏÂ£¬ÐèÒª¸üÐÂ¾ÉµÄ first_free_page_no µÄ next_free_page_no Ö¸ÏòÐÂÒ³Ãæ¡£
        // µ«ÕâÐèÒª¶îÍâ»ñÈ¡¾ÉµÄ first_free_page_no ¶ÔÓ¦µÄ page_handle£¬»áÔö¼Ó¸´ÔÓÐÔ¡£
        // ×î¼òµ¥µÄ²ßÂÔÊÇ£¬Èç¹ûËüÓÐ¿ÕÏÐ²ÛÎ»£¬¾ÍÈÃËü³ÉÎªÐÂµÄ first_free_page_no¡£
        // ²¢ÇÒ½«Æä next_free_page_no Ö¸Ïò¾ÉµÄ first_free_page_no¡£
        new_page_handle.page_hdr->next_free_page_no = file_hdr_.first_free_page_no;
        file_hdr_.first_free_page_no = new_page_id.page_no;
    }

    // ±ê¼Ç file_hdr ÎªÔà£¬ÒòÎª num_pages ºÍ first_free_page_no ¿ÉÄÜÒÑ¸üÐÂ
    // file_hdr µÄ Flush Ó¦¸ÃÓÉ RmManager ¸ºÔðÔÚ close_file Ê±½øÐÐ¡£
    // ÕâÀïÎÒÃÇÖ»ÐèÒªÈ·±£ page ±ê¼ÇÎªÔà¡£
    // new_page ÒÑ¾­±» NewPage ±ê¼ÇÎª pin_count=1, is_dirty=false¡£
    // ÔÚÕâÀï£¬ÎÒÃÇ²»ÐèÒªÁ¢¼´½«Æä±ê¼ÇÎªÔà£¬ÒòÎªËüµÄÊý¾Ý»¹Ã»ÓÐ±»Ð´Èë¡£
    // Ö»ÓÐµ± insert_record Êµ¼ÊÐ´ÈëÊý¾Ýºó£¬²Å½«Æä±ê¼ÇÎªÔà¡£

    return new_page_handle;
}

/**
 * @brief ´´½¨»ò»ñÈ¡Ò»¸ö¿ÕÏÐµÄpage handle
 *
 * @return RmPageHandle ·µ»ØÉú³ÉµÄ¿ÕÏÐpage handle
 * @note pin the page, remember to unpin it outside!
 */
RmPageHandle RmFileHandle::create_page_handle() {
    // Todo:
    // 1. ÅÐ¶Ïfile_hdr_ÖÐÊÇ·ñ»¹ÓÐ¿ÕÏÐÒ³
    //     1.1 Ã»ÓÐ¿ÕÏÐÒ³£ºÊ¹ÓÃ»º³å³ØÀ´´´½¨Ò»¸öÐÂpage£»¿ÉÖ±½Óµ÷ÓÃcreate_new_page_handle()
    //     1.2 ÓÐ¿ÕÏÐÒ³£ºÖ±½Ó»ñÈ¡µÚÒ»¸ö¿ÕÏÐÒ³
    // 2. Éú³Épage handle²¢·µ»Ø¸øÉÏ²ã

    if (file_hdr_.first_free_page_no != RM_NO_PAGE) {
        // 1.2 ÓÐ¿ÕÏÐÒ³£ºÖ±½Ó»ñÈ¡µÚÒ»¸ö¿ÕÏÐÒ³
        int free_page_no = file_hdr_.first_free_page_no;
        RmPageHandle page_handle = fetch_page_handle(free_page_no);
        // ¸üÐÂ file_hdr_.first_free_page_no Ö¸ÏòÏÂÒ»¸ö¿ÕÏÐÒ³
        file_hdr_.first_free_page_no = page_handle.page_hdr->next_free_page_no;
        return page_handle;
    } else {
        // 1.1 Ã»ÓÐ¿ÕÏÐÒ³£ºÊ¹ÓÃ»º³å³ØÀ´´´½¨Ò»¸öÐÂpage£»¿ÉÖ±½Óµ÷ÓÃcreate_new_page_handle()
        return create_new_page_handle();
    }
}

/**
 * @brief µ±page handleÖÐµÄpage´ÓÒÑÂú±ä³ÉÎ´ÂúµÄÊ±ºòµ÷ÓÃ
 *
 * @param page_handle
 * @note only used in delete_record()
 */
void RmFileHandle::release_page_handle(RmPageHandle &page_handle) {
    // Todo:
    // µ±page´ÓÒÑÂú±ä³ÉÎ´Âú£¬¿¼ÂÇÈçºÎ¸üÐÂ£º
    // 1. page_handle.page_hdr->next_free_page_no
    // 2. file_hdr_.first_free_page_no

    // Ö»ÓÐµ±Ò³Ãæ´ÓÂú±äÎª²»ÂúÊ±²ÅÐèÒª´¦Àí
    // È·±£¸ÃÒ³Ãæµ±Ç°²»ÊÇÂúµÄ£¬ÇÒÖ®Ç°ÊÇÂúµÄ£¨Í¨¹ý num_records ÅÐ¶Ï£©
    // ²¢ÇÒËü²»ÔÚ¿ÕÏÐÁ´±íÖÐ
    if (page_handle.page_hdr->num_records < file_hdr_.num_records_per_page &&
        page_handle.page_hdr->next_free_page_no == RM_NO_PAGE && // È·±£ËüÖ®Ç°ÊÇÂúµÄ»òÕß²»ÔÚÁ´±íÖÐ
        file_hdr_.first_free_page_no != page_handle.page->GetPageId().page_no) // È·±£Ëü²»ÊÇµ±Ç°µÄµÚÒ»¸ö¿ÕÏÐÒ³
    {
        // ½«¸ÃÒ³ÃæÌí¼Óµ½¿ÕÏÐÒ³Á´±íµÄÍ·²¿
        page_handle.page_hdr->next_free_page_no = file_hdr_.first_free_page_no;
        file_hdr_.first_free_page_no = page_handle.page->GetPageId().page_no;

        // ±ê¼ÇÒ³ÃæÎªÔà£¬ÒòÎª page_hdr->next_free_page_no ¸Ä±äÁË
        // ±ê¼Ç file_hdr ÎªÔà£¬ÒòÎª first_free_page_no ¸Ä±äÁË
        // ÕâÐ©ÔàÒ³»áÔÚ FlushPage »ò FlushAllPages Ê±Ð´Èë´ÅÅÌ
        // ÕâÀïÖ»ÊÇ±ê¼Ç£¬¾ßÌåµÄ Flush ²Ù×÷ÓÉ BufferPoolManager ¸ºÔð
        // page_handle.page->is_dirty_ = true; // Ò³ÃæÄÚÈÝÒÑÐÞ¸Ä£¬±ê¼ÇÎªÔà
    }
    // ±ê¼ÇÒ³ÃæÎªÔà£¬ÒòÎªÆäÔªÊý¾Ý£¨num_records£©¿ÉÄÜÒÑ¸Ä±ä
    //page_handle.page->is_dirty_ = true;
}

// used for recovery (lab4)
// Õâ¸öº¯ÊýÔÚÄãµÄÔ­Ê¼ÎÄ¼þÖÐÓÐ£¬µ«Óë insert_record(char *buf, Context *context) ÖØÔØ£¬
// ÇÒÂß¼­ÉÏ¿ÉÄÜÓÃÓÚ»Ö¸´¡£ÕâÀï±£ÁôÔ­Ñù£¬µ«Òª×¢Òâº¯ÊýÇ©Ãû³åÍ»¡£
// Èç¹ûËüÖ»ÊÇÓÃÓÚ»Ö¸´£¬ÇÒ²»×÷Îª³£¹æ²åÈë£¬¿ÉÒÔ¿¼ÂÇÖØÃüÃû»òµ÷Õû¿É¼ûÐÔ¡£
void RmFileHandle::insert_record(const Rid &rid, char *buf) {
    // Õâ¸öº¯ÊýËÆºõÊÇÓÃÓÚ»Ö¸´µÄ£¬ÓëÉÏÃæµÄ insert_record(char *buf, Context *context) Âß¼­²»Í¬¡£
    // Ëü¼ÙÉè rid ÒÑ¾­È·¶¨£¬²¢Ö±½Ó²Ù×÷¡£
    // È·±£ page_no ÔÚ·¶Î§ÄÚ£¬Èç¹û²»ÔÚ£¬ÔòÐèÒª´´½¨ÐÂÒ³
    if (rid.page_no >= file_hdr_.num_pages) {
        // ÕâÀïÂß¼­¿ÉÄÜÐèÒªµ÷Õû£¬Èç¹û rid.page_no ³¬³öµ±Ç°ÎÄ¼þÒ³Êý£¬
        // Ó¦¸Ã´´½¨×ã¹»µÄÒ³ÃæÖ±µ½ rid.page_no ´æÔÚ¡£
        // ¼òµ¥Æð¼û£¬ÕâÀï¼ÙÉèËüÖ»ÊÇÒ»¸ö¶ÏÑÔ»òÒ»¸öÐèÒª´¦ÀíµÄ´íÎó¡£
        // »òÕß£¬Èç¹ûÕâ¸öº¯ÊýÖ»ÓÃÓÚ»Ö¸´£¬ÄÇÃ´Ò³ÃæÓ¦¸ÃÒÑ¾­´æÔÚ¡£
        // ÎªÁË±ÜÃâ±àÒë´íÎó£¬ÎÒÃÇÔÝÊ±²»ÔÚÕâÀïµ÷ÓÃ create_new_page_handle()£¬
        // ÒòÎªËü»á¸Ä±ä file_hdr_.num_pages£¬¿ÉÄÜÓë»Ö¸´Âß¼­³åÍ»¡£
        // Èç¹ûÐèÒª´´½¨ÐÂÒ³£¬Ó¦¸ÃÔÚµ÷ÓÃ´Ëº¯ÊýÖ®Ç°Íê³É¡£
        throw InternalError("Page " + std::to_string(rid.page_no) + " does not exist for recovery insert.");
    }
    RmPageHandle pageHandle = fetch_page_handle(rid.page_no);
    Bitmap::set(pageHandle.bitmap, rid.slot_no);
    pageHandle.page_hdr->num_records++;
    if (pageHandle.page_hdr->num_records == file_hdr_.num_records_per_page) {
        // Ò³ÃæÒÑÂú£¬´Ó¿ÕÏÐÁ´±íÖÐÒÆ³ý
        // ÕâÀïµÄÂß¼­ÐèÒªÈ·±£¸ÃÒ³ÃæÖ®Ç°ÔÚ¿ÕÏÐÁ´±íÖÐ
        // ²¢ÇÒ¸üÐÂ file_hdr_.first_free_page_no
        if (file_hdr_.first_free_page_no == pageHandle.page->GetPageId().page_no) {
            file_hdr_.first_free_page_no = pageHandle.page_hdr->next_free_page_no;
        }
        pageHandle.page_hdr->next_free_page_no = RM_NO_PAGE; // ±ê¼Ç¸ÃÒ³ÒÑÂú
    }

    char *slot = pageHandle.get_slot(rid.slot_no);
    memcpy(slot, buf, file_hdr_.record_size);

    // ±ê¼ÇÒ³ÃæÎªÔà²¢È¡Ïû¹Ì¶¨
    MarkDirtyAndUnpin(buffer_pool_manager_, pageHandle.page, true);
}

